package io.mft.datasource.mq.container;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import cn.hutool.core.util.ObjectUtil;
import io.mft.commons.config.mq.MqConsumer;
import io.mft.commons.config.mq.MqExchange;
import io.mft.commons.config.mq.MqQueue;
import io.mft.commons.exception.MFTException;
import io.mft.commons.function.response.kv.KVMap;
import io.mft.datasource.mq.rabbitmq.MqDatasource;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class MqContainer {
	protected  Map<MqContainerType, Map<String, MqDatasource>> mapMqDatasource = new ConcurrentHashMap<>();
	protected Map<MqContainerType, Map<String,Map<String, MqQueue>>> mapMqQueue = new ConcurrentHashMap<>();
	protected Map<MqContainerType, Map<String,Map<String, MqExchange>>> mapMqExchange = new ConcurrentHashMap<>();
	protected Map<MqContainerType, Map<String,Map<String, MqConsumer>>> mapMqConsumer = new ConcurrentHashMap<>();

	public enum MqDatasourceType {
		producer, consumer, admin;
	}

	{
		Map<String, MqDatasource> dMap=new ConcurrentHashMap<String, MqDatasource>();
		mapMqDatasource.put(mqContainerType(), dMap);
		Map<String,Map<String, MqExchange>> map = new ConcurrentHashMap<>();
		mapMqExchange.put(mqContainerType(), map);
		Map<String,Map<String, MqQueue>> qmap = new ConcurrentHashMap<>();
		mapMqQueue.put(mqContainerType(), qmap);
		
		Map<String,Map<String, MqConsumer>> cmap = new ConcurrentHashMap<>();
		mapMqConsumer.put(mqContainerType(), cmap);
	}
	public abstract void shutDown(String datasourceId);

	private void initMqDatasource(MqDatasource datasource) {
		log.warn("开始初始化MQ[{}]数据源:{}-{}",this.getClass().getSimpleName(),datasource.getId(),datasource.getDesc());
		if (mapMqDatasource.get(mqContainerType()).containsKey(datasource.getId())) {
			MqDatasource mqDatasource = mapMqDatasource.get(mqContainerType()).get(datasource.getId());
			if (mqDatasource.uniqueId().equals(datasource.uniqueId())) {
				return;
			} else {
				shutDown(datasource.getId());
			}
		}
		mapMqDatasource.get(mqContainerType()).put(datasource.getId(), datasource);
		initedMqDatasource(datasource);
		log.warn("初始化完成MQ[{}]数据源:{}-{}",this.getClass().getSimpleName(),datasource.getId(),datasource.getDesc());
	}

    public void AddMqDatasource(MqDatasource datasource) {
		initMqDatasource(datasource);
	}
    
    public void addMqQueue(String pluginId,MqQueue queue) {
    	Map<String,Map<String, MqQueue>> map=mapMqQueue.get(mqContainerType());
    	Map<String, MqQueue> pluginMapMqQueue=map.get(pluginId);
    	if (ObjectUtil.isNull(pluginMapMqQueue)) {
    		pluginMapMqQueue=new ConcurrentHashMap<>();
    		map.put(pluginId, pluginMapMqQueue);
		}
    	if (!pluginMapMqQueue.containsKey(queue.getId())) {
    		pluginMapMqQueue.put(queue.getId(), queue);
			addedMqQueue(pluginId,queue);
			log.warn("初始化完成MQ队列->>>系统插件:{}->>>数据源:{}->>>ID:{}->>>队列名称:{}->>>备注:{}",pluginId,queue.getDatasourceId(),queue.getId(),queue.getName(),queue.getNote());
		}

    	
	}
    
    public void addMqConsumer(String pluginId,MqConsumer consumer) {
    	//Map<String, MqConsumer> map=mapMqConsumer.get(mqContainerType());
    	Map<String,Map<String, MqConsumer>> map=mapMqConsumer.get(mqContainerType());
    	Map<String, MqConsumer> pluginMapMqConsumer=map.get(pluginId);
    	if (ObjectUtil.isNull(pluginMapMqConsumer)) {
    		pluginMapMqConsumer=new ConcurrentHashMap<>();
    		map.put(pluginId, pluginMapMqConsumer);
		}
    	if (!pluginMapMqConsumer.containsKey(consumer.getId())) {
    		pluginMapMqConsumer.put(consumer.getId(), consumer);
			addedMqConsumer(pluginId,consumer);
			log.warn("初始化完成MQ消费者->>>系统插件:{}->>>数据源:{}->>>ID:{}->>>队列:{}->>>备注:{}",pluginId,consumer.getDatasourceId(),consumer.getId(),consumer.getQueueId(),consumer.getNote());
		}
	}
    
    
    public void addMqExchange(String pluginId,MqExchange exchange) {
    	//Map<String, MqExchange> map=mapMqExchange.get(mqContainerType());
    	Map<String,Map<String, MqExchange>> map=mapMqExchange.get(mqContainerType());
    	Map<String, MqExchange> pluginMapMqExchange=map.get(pluginId);
    	if (ObjectUtil.isNull(pluginMapMqExchange)) {
    		pluginMapMqExchange=new ConcurrentHashMap<>();
    		map.put(pluginId, pluginMapMqExchange);
		}
    	if (!pluginMapMqExchange.containsKey(exchange.getId())) {
    		pluginMapMqExchange.put(exchange.getId(), exchange);
        	addedMqExchange(pluginId,exchange);
        	log.warn("初始化完成MQ交换机/topic->>>系统插件:{}->>>数据源:{}->>>ID:{}->>>备注:{}",pluginId,exchange.getDatasourceId(),exchange.getId(),exchange.getNote());
		}

	}
    abstract protected void addedMqExchange(String pluginId,MqExchange exchange);
    abstract protected void addedMqQueue(String pluginId,MqQueue queue);
    abstract protected void addedMqConsumer(String pluginId,MqConsumer consumer);
	abstract protected void initedMqDatasource(MqDatasource datasource);
	/**
	 * 发送消息
	 * @说明: TODO
	 * @方法名称: sendMessage
	 * @参数 @param exchangeId
	 * @参数 @param message
	 * @返回类型 void    
	 * @创建者: 胡常建  
	 * @创建时间: 2021年4月26日 下午1:24:02
	 * @修改者: 胡常建  
	 * @修改时间: 2021年4月26日 下午1:24:02
	 */
	abstract public  void sendMessage(String pluginId,String exchangeId,KVMap message);
	
	abstract public MqContainerType mqContainerType();
	
	public MqQueue getQueue(String pluginId,String queueId) {
		Map<String,Map<String, MqQueue>> pluginMapMqQueue= mapMqQueue.get(mqContainerType());
		Map<String, MqQueue> map=pluginMapMqQueue.get(pluginId);
		if (ObjectUtil.isNull(pluginMapMqQueue)) {
			throw new MFTException(-1, "系统插件["+pluginId+"]不存在队列配置信息,请检查!");
		}
		return map.get(queueId);
	}
}
